package Autodesk;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

public class SessionFilterCounter extends Configured implements Tool{
	public Path prodPath;
	public static class LogMapper extends Mapper<Object, Text, Text, IntWritable> {
		
		// private final static IntWritable one = new IntWritable(1);
		private Text word = new Text();
		private IntWritable one= new IntWritable(1);
		Hashtable<String, String> ht = new Hashtable<String, String>();
		private Map<String,String> StaticListTime=new HashMap<String,String>();

		public void setup(Context context) {

			Configuration conf = context.getConfiguration();
			if (conf.getBoolean("time.patterns", false)) {
				Path[] patternsFiles = new Path[0];
				try {
					patternsFiles = DistributedCache.getLocalCacheFiles(conf);
				} catch (IOException ioe) {
					System.err
							.println("Caught exception while getting cached files: ");
				}
				for (Path patternsFile : patternsFiles) {
						getStaticList(patternsFile, StaticListTime);

				}

			}
		}
		private void getStaticList(Path patternFile,Map<String,String> staticList){
			try{
				BufferedReader fis=new BufferedReader(new FileReader(patternFile.toString()));	
				String pattern=null;
				
				while((pattern =fis.readLine())!=null){ //We consider that the input mapping product document is: ID-ProdAb/1/2/3/4/5 - ReleaseAb
					String record[] = pattern.split("\t");

						String id=record[0];
						String hour=record[1];
						String minute=record[2];
						staticList.put(hour+":"+minute,id);
				}
				fis.close();
			}catch(IOException ioe){
		         System.err.println("Caught exception while parsing the cached file '");
			}
		}
		class SessionNullException extends Exception{
			public SessionNullException (){
				System.out.println("Cant' find the ID. Please check the Mapping Document.");
			}
			public SessionNullException (String field){
				System.out.println("Cant' find the field: "+field+". Check the Mapping Document please.");
			}
		}
		
		private String lookForKey(Map<String,String> staticList,String key) throws SessionNullException{
			String id=null;
			String[] keySp=key.split(":");
			if(Integer.parseInt(keySp[0])<10){
				keySp[0]="0"+Integer.parseInt(keySp[0]);
			}
			String hourSplited=keySp[0]+":"+keySp[1];
			id=staticList.get(hourSplited);
			if(id==null){
				throw new SessionNullException(key);
			}
			return id;
		}
	public void map(Object key, Text value, Context context)
			throws IOException, InterruptedException {

		String record[] = value.toString().split("\t");
		StringBuffer bf = new StringBuffer();
		bf.append(record[0]);// TimeDimID
		
		bf.append("\t");
		bf.append(record[2]);// LanguageID
		bf.append("\t");
		bf.append(record[3]);// ProductID
		bf.append("\t");
		bf.append(record[4]);// ChannelID
		bf.append("\t");
		try {
			bf.append(lookForKey(StaticListTime, record[5]));
			bf.append("\t");
			bf.append(lookForKey(StaticListTime, record[6]));
			bf.append("\t");
			bf.append(record[7]);// Country
			bf.append("\t");
			bf.append(record[1]);// IP		
			word.set(bf.toString());
		} catch (SessionNullException e) {
			e.printStackTrace();
		}
		context.write(word, one);
	}
}



public static class LogReducer extends
	Reducer<Text, IntWritable, Text, IntWritable> {

	private IntWritable result = new IntWritable();
	private MultipleOutputs<Text,IntWritable> mos;
	 
	 public void setup(Context context) {
		 mos = new MultipleOutputs<Text,IntWritable>(context);
	 }
	
	public void reduce(Text key, Iterable<IntWritable> values,
		Context context) throws IOException, InterruptedException {
	
	int sum = 0;
	for (IntWritable val : values) {
		sum += val.get();
	}
	result.set(sum);
	
	context.write(key, result);
	String[] keySplitted=key.toString().split("\t",2);
	mos.write(key, result, keySplitted[0]+"_SessionReport");

	}
	 public void cleanup(Context context) throws IOException, InterruptedException {
		 mos.close();
	 }
}

public int run(String[] args) throws Exception { 
	Configuration conf = getConf();
	
	Job job = new Job(conf, "SessionFilter2");
	job.setJarByClass(Autodesk.SessionFilter.class);

	job.setMapperClass(LogMapper.class);
	job.setReducerClass(LogReducer.class);
	
	job.setMapOutputKeyClass(Text.class);
	job.setMapOutputValueClass(IntWritable.class);
	
	job.setOutputKeyClass(Text.class);
	job.setOutputValueClass(IntWritable.class);

    //FileInputFormat.setInputPaths(job, new Path(args[0]));
    //FileOutputFormat.setOutputPath(job, new Path(args[1]));
    
	List<String> other_args = new ArrayList<String>();
	for (int i = 0; i < args.length; ++i) {
		if ("-dayTime".equals(args[i])) {
			prodPath=new Path(args[++i]);
			DistributedCache.addCacheFile(prodPath.toUri(),
					job.getConfiguration());
			job.getConfiguration().setBoolean("time.patterns",
					true);
			job.getConfiguration().set("time.path", prodPath.getName());

		}else{
			other_args.add(args[i]);
		}
	}
    FileInputFormat.setInputPaths(job, new Path(other_args.get(0)));
    FileOutputFormat.setOutputPath(job, new Path(other_args.get(1)));
    
    MultipleOutputs.addNamedOutput(job,"SessionReport", TextOutputFormat.class, Text.class, IntWritable.class);
    System.exit(job.waitForCompletion(true)?0:1);
	
	return 0;

}
public static void main(String[] args) throws Exception {
	int res = ToolRunner.run(new Configuration(), new SessionFilterCounter(), args);
	System.exit(res);
	}


}
